package com.gutsyzhan.bilibili.config;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.gutsyzhan.bilibili.constant.UserMomentsConstant;
import com.gutsyzhan.bilibili.domain.UserFollowing;
import com.gutsyzhan.bilibili.domain.UserMoments;
import com.gutsyzhan.bilibili.service.UserFollowingService;
import com.mysql.jdbc.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

@Configuration
public class RocketMQConfig {
    @Value("${rocketmq.name.server.address}")
    private String nameServerAddress;

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private UserFollowingService userFollowingService;

    @Bean("momentsProducer")
    public DefaultMQProducer momentsProducer() throws Exception {
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer(UserMomentsConstant.GROUP_MOMENTS);
        // 设置NameServer地址
        defaultMQProducer.setNamesrvAddr(nameServerAddress);
        //启动生产者
        defaultMQProducer.start();
        return defaultMQProducer;
    }

    /**
     * 采用推方式
     */
    @Bean("momentsConsumer")
    public DefaultMQPushConsumer momentsConsumer() throws Exception {
        // 实例化消费者
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(UserMomentsConstant.GROUP_MOMENTS);
        // 设置NameServer地址
        defaultMQPushConsumer.setNamesrvAddr(nameServerAddress);
        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        defaultMQPushConsumer.subscribe(UserMomentsConstant.TOPIC_MOMENTS, "*");
        //注册消息监听器，注册回调实现类来处理从broker拉取回来的消息
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //由于采用同步发送方式，因此这个消息列表中实际上只有一条消息
                MessageExt messageExt = msgs.get(0);
                if(messageExt == null){
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                String msg = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                UserMoments userMoments = JSONObject.toJavaObject(JSON.parseObject(msg), UserMoments.class);
                Long userId = userMoments.getUserId();
                //找到该用户的所有粉丝
                List<UserFollowing> userFans = userFollowingService.getUserFans(userId);
                userFans.forEach(userFan->{
                    //将动态写入Redis中，注意是给每个粉丝写入对应的订阅缓存中，采用字符串这一数据结构
                    String key = "subscribed-" + userFan.getUserId();
                    //取出当前粉丝目前已订阅的动态字符串
                    String subscribedStr = redisTemplate.opsForValue().get(key);
                    List<UserMoments> subscribedList;
                    if(StringUtils.isNullOrEmpty(subscribedStr)){
                        //已订阅的动态为空
                        subscribedList = new ArrayList<>();
                    }else{
                        //将已订阅的动态字符串转换为列表
                        subscribedList = JSONArray.parseArray(subscribedStr, UserMoments.class);
                    }
                    //将当前推过来的动态放入该列表中
                    subscribedList.add(userMoments);
                    //将该列表转换为字符串
                    String subscribedListStr = JSONObject.toJSONString(subscribedList);
                    //将该字符串放入Redis中
                    redisTemplate.opsForValue().set(key, subscribedListStr);
                });
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //启动消费者
        defaultMQPushConsumer.start();
        return defaultMQPushConsumer;
    }
}
